/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.handler.msg;

import cn.sunny360.core.common.PhResult;
import cn.sunny360.core.handler.eb.EbConsumerHandler;
import cn.sunny360.core.service.UserService;
import cn.sunny360.core.utils.Container;
import cn.sunny360.messsage.MsgMessage;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.core.shareddata.AsyncMap;
import io.vertx.core.shareddata.SharedData;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;

public class ConnectionAction implements MsgAction {

    private static Logger logger = LoggerFactory.getLogger(ConnectionAction.class);

    private EbConsumerHandler ebConsumerHandler;
    private UserService userService;

    public ConnectionAction() {
        ebConsumerHandler = new EbConsumerHandler();
        userService = Container.getComponent(UserService.class);
    }
    @Override
    public Promise<PhResult> handler(Vertx vertx, ServerWebSocket webSocket, MsgMessage message) {
        Promise<PhResult> promise = Promise.promise();
        //从消息中取用户名
        String userId =  message.getFrom();
        //从消息中取token
        String token = message.getToken();
        logger.debug("用户id：" + userId + "\ttoken:" + token);
        if (StringUtils.isBlank(userId) || StringUtils.isBlank(token)) {
            promise.fail("message format error!");
            return promise;
        }

        //更新在线用户列表
        Promise<Boolean> connPromise = Promise.promise();
        userService.wsConnection(Long.valueOf(userId), token, connPromise);
        // 保存websocket和token的关系
        Promise<PhResult> mapPromise = saveAsyncMap(vertx, "ws_token_map", webSocket.binaryHandlerID(), token);
        CompositeFuture.all(
                        connPromise.future(),
                        mapPromise.future()
                ).onFailure(promise::fail)
                .onSuccess(f->{
                    //  eventBus注册监听
                    ebConsumerHandler.eventBusConsumer(vertx.eventBus(), webSocket, token);
                    //返回连接成功消息
                    MsgMessage msgSuccess = MsgMessage.newInstance()
                            .setType("success")
                            .setVersion(0)
                            .setTime(new DateTime().toString("yyyy-MM-dd HH:mm:ss"));
                    webSocket.write(Buffer.buffer(msgSuccess.toByteArray()));
                    promise.complete(PhResult.ok());
                });

        return promise;
    }

    private Promise<PhResult> saveAsyncMap(Vertx vertx, String mapKey, String key, String vlaue) {
        Promise<PhResult> promise = Promise.promise();
        SharedData sd = vertx.sharedData();
        Future<AsyncMap<String, String>> mapFuture = null;
        if (vertx.isClustered()) {
            mapFuture = sd.getClusterWideMap(mapKey);
        } else {
            mapFuture = sd.getAsyncMap(mapKey);
        }
        mapFuture.onSuccess(asyncMap->{
            asyncMap.put(key, vlaue, h->{
                if (h.succeeded()) {
                    promise.complete(PhResult.ok());
                } else {
                    promise.fail(h.cause());
                }
            });
        }).onFailure(promise::fail);
        return promise;
    }
}
